
package cn.qtone.test.pinpoint;

import cn.qtone.qtframe.common.utils.DateUtils;
import cn.qtone.qtframe.data.hbase.client.HBaseAdminTemplate;
import cn.qtone.qtframe.data.hbase.client.HbaseOperations2;
import cn.qtone.qtframe.data.hbase.client.HbaseTemplate2;
import cn.qtone.qtframe.data.hbase.client.PooledHTableFactory;
import cn.qtone.qtframe.data.hbase.client.RowMapper;
import cn.qtone.qtframe.data.hbase.client.distributor.RangeOneByteSimpleHash;
import cn.qtone.qtframe.threadpool.CommonThreadPoolManager;
import com.google.common.base.Stopwatch;
import com.navercorp.pinpoint.common.server.bo.SpanBo;
import com.navercorp.pinpoint.common.server.bo.serializer.RowKeyDecoder;
import com.navercorp.pinpoint.common.server.bo.serializer.trace.v2.TraceRowKeyDecoderV2;
import com.navercorp.pinpoint.common.util.TransactionId;
import com.navercorp.pinpoint.web.mapper.SpanMapperV2;
import com.navercorp.pinpoint.web.vo.Range;
import com.navercorp.pinpoint.web.vo.scatter.Dot;
import com.sematext.hbase.wd.RowKeyDistributorByHashPrefix;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Created by tanghc on 17/2/20.
 */
public class HbaseClient {

	// 声明静态配置
	static Configuration conf = null;
	static HbaseOperations2 template = null;
	static HBaseAdminTemplate adminTemplate = null;
	final static ConcurrentHashMap<String,Integer> map = new ConcurrentHashMap<>();
	static {
		conf = HBaseConfiguration.create();
//		conf.set("hbase.zookeeper.quorum", "localhost");
//		conf.set("hbase.zookeeper.quorum", "172.31.64.11");
		conf.set("hbase.zookeeper.quorum", "10.254.141.132");
		conf.set("hbase.zookeeper.property.clientPort", "2181");
		conf.setLong(HConstants.HBASE_REGIONSERVER_LEASE_PERIOD_KEY, 1200000);
		HbaseTemplate2 hbaseTemplate2 = new HbaseTemplate2();
		hbaseTemplate2.setConfiguration(conf);
		hbaseTemplate2.setTableFactory(new PooledHTableFactory(conf));
		hbaseTemplate2.afterPropertiesSet();
		template = hbaseTemplate2;
		adminTemplate = new HBaseAdminTemplate(conf);
	}
	static HbaseOperations2 newHbaseOperations2(){
		HbaseTemplate2 hbaseTemplate2 = new HbaseTemplate2();
		hbaseTemplate2.setConfiguration(conf);
		hbaseTemplate2.setTableFactory(new PooledHTableFactory(conf));
		hbaseTemplate2.afterPropertiesSet();
		return hbaseTemplate2;
	}

	public static void main(String[] args)
			throws Exception {

		// 创建表
		String tableName = "TraceV2";
//		String tableName = "blog2";
		final TableName table = TableName.valueOf(tableName);
		String[] family = {
				"article", "author"
		};
//		template.execute(table, new TableCallback<Object>() {
//			@Override
//			public Object doInTable(Table table) throws Throwable {
//				return null;
//			}
//		})
//		 creatTable(tableName, family);

		// 为表添加数据

		String[] column1 = {
				"title", "content", "tag"
		};
		String[] value1 = {
				"Head First HBase",
				"HBase is the Hadoop database. Use it when you need random, realtime read/write access to your Big Data.",
				"Hadoop,HBase,NoSQL"
		};
		String[] column2 = {
				"name", "nickname"
		};
		String[] value2 = {
				"nicholas", "lee"
		};
//		template.put(table,"title".getBytes(),"article".getBytes(),"rowkey001".getBytes(),"tile2".getBytes());
//		addData("rowkey1", "blog2", column1, value1, column2, value2);
//		template.put(table, Bytes.toBytes("content"), Bytes.toBytes("article"), Bytes.toBytes("rowkey002"), Bytes.toBytes("conte1"));
//		template.put(table,  Bytes.toBytes("rowkey002"), Bytes.toBytes("article"),Bytes.toBytes("content"), Bytes.toBytes("conte1"));
//		template.delete(table, new Delete(Bytes.toBytes("title")));
//		addData("rowkey3", "blog2", column1, value1, column2, value2);

//		// 遍历查询
//		getResultScann("blog2", "rowkey4", "rowkey5");
//		// 根据row key范围遍历查询
//		getResultScann("blog2", "rowkey4", "rowkey5");
//
//		// 查询
//		getResult("blog2", "rowkey1");
//
//		// 查询某一列的值
//		getResultByColumn("blog2", "rowkey1", "author", "name");
//
//		// 更新列
//		updateTable("blog2", "rowkey1", "author", "name", "bin");
//
//		// 查询某一列的值
//		getResultByColumn("blog2", "rowkey1", "author", "name");
//
//		// 查询某列的多版本
//		getResultByVersion("blog2", "rowkey1", "author", "name");

//		// 删除一列
//		deleteColumn("blog2", "rowkey1", "author", "nickname");
//
//		// 删除所有列
//		deleteAllColumn("blog2", "rowkey1");
//
//		// 删除表
//		deleteTable("blog2");
		final Scan scan = new Scan();
//		template.find(table,)
//		HColumnDescriptor[] columnFamilies = adminTemplate.getAdmin().getTableDescriptor(table).getColumnFamilies();
//		for(HColumnDescriptor columnDescriptor: columnFamilies){
//			System.out.println("======"+columnDescriptor.getNameAsString());
//		}
//		scan.setStartRow(Bytes.toBytes(1));
//		scan.setStopRow(Bytes.toBytes(1000));


//		scan.addColumn(fam, null);

//		scan.setMaxVersions();
		final long start = DateUtils.parse("2017-02-20 09:00:00", "yyyy-MM-dd HH:mm:ss").getTime();
		final long end = DateUtils.parse("2017-02-20 18:00:00", "yyyy-MM-dd HH:mm:ss").getTime();
//		scan.setColumnFamilyTimeRange(fam,start,end);

		final AtomicLong count;
//		final long step = 1*60*60*1000;
		final int threadCount = 1;
		final long step = (end - start) / threadCount;
		count = new AtomicLong(0);
		Stopwatch stopwatch = Stopwatch.createStarted();
		final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
		for(int i=0;i<threadCount;i++){
			final int j = i;

			CommonThreadPoolManager.getInstance().addThread(new Runnable() {
				@Override
				public void run() {
					try {
						final Scan scan = new Scan();
						byte[] fam = Bytes.toBytes("S");
						scan.addFamily(fam);
						scan.setCaching(256);
						scan.setBatch(1000);
						scan.setMaxResultSize(5000);
						try {
							long s = start + (j * step ) + j;
							long e = s + step;
							if (j == threadCount - 1) {
								e = end;
							}
							scan.setTimeRange(s, e);
						} catch (IOException e) {
							e.printStackTrace();
						}
						process(table, scan, count, map);
					} catch (Exception e) {
						e.printStackTrace();
					} finally {
						System.out.println("==================finish" + countDownLatch.getCount());
						countDownLatch.countDown();
//						write(j,mapi);
					}
				}
			});
		}
		countDownLatch.await();
		CommonThreadPoolManager.getInstance().shutdown();
		System.out.println("times=========" + stopwatch.elapsed(TimeUnit.SECONDS));
		System.out.println("count========="+count.longValue());
		System.out.println("XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX");
		write(-1,map);
//		System.out.println(StringUtils.join(adminTemplate.getAdmin().listTableNames(),"###"));
	}
	private static void write(int i,Map<String,Integer> map){
		try {
			File file = new File("/tmp/rpc"+i+".txt");
			File fileOther = new File("/tmp/rpc-other"+i+".txt");
			FileWriter fileWriter = new FileWriter(file);
			FileWriter fileOtherWriter = new FileWriter(fileOther);
			for(Map.Entry<String,Integer> entry: map.entrySet()){
				if(entry.getKey().indexOf(":") != -1){
					fileWriter.write(entry.getKey()+"\t"+entry.getValue());
					fileWriter.write("\n");
				}else{
					fileOtherWriter.write(entry.getKey()+"\t"+entry.getValue());
					fileOtherWriter.write("\n");
				}
			}
			fileWriter.flush();
			fileOtherWriter.flush();
		} catch (IOException e) {
			e.printStackTrace();
		}

	}
	private static void count(String key,Map<String,Integer> map){
		if(null == key){
			return ;
		}
		Integer count = map.get(key);
		if(count == null){
			count = 0;
		}
		count ++;
		map.put(key,count);
	}
	private static void process(TableName table,Scan scan,final AtomicLong count,final Map<String,Integer> map){
		try {
			RangeOneByteSimpleHash oneByteSimpleHash = new RangeOneByteSimpleHash(32,40,256);
			RowKeyDistributorByHashPrefix rowKeyDistributorByHashPrefix = new RowKeyDistributorByHashPrefix(oneByteSimpleHash);
			RowKeyDecoder rowKeyDecoder = new TraceRowKeyDecoderV2();
			final SpanMapperV2 spanMapperV2 = new SpanMapperV2(rowKeyDecoder);
			template.findParallel(table, scan, rowKeyDistributorByHashPrefix, new RowMapper<Object>() {
				@Override
				public Object mapRow(Result result, int rowNum) throws Exception {
					try {
						List<SpanBo> spanBos = spanMapperV2.mapRow(result, rowNum);
						System.out.println("===============" + rowNum);
						if (null != spanBos) {
							for (SpanBo spanBo : spanBos) {
								System.out.println("########" + spanBo.getRpc());
								count(spanBo.getRpc(), map);
								long incrementAndGet = count.incrementAndGet();
//								if(count.incrementAndGet() > 10000){
//									return null;
//								}
							}
						}
					} catch (Exception e) {
						e.printStackTrace();
					}
					return null;
				}
			}, 8);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	private static void process(TableName table,Scan scan,final AtomicLong count,final Map<String,Integer> map,TransactionId transactionId){
		try {

			RowKeyDecoder rowKeyDecoder = new TraceRowKeyDecoderV2();
			final SpanMapperV2 spanMapperV2 = new SpanMapperV2(rowKeyDecoder);
			template.find(table, scan,new RowMapper<Object>() {
				@Override
				public Object mapRow(Result result, int rowNum) throws Exception {
					try {
						List<SpanBo> spanBos = spanMapperV2.mapRow(result, rowNum);
						System.out.println("===============" + rowNum);
						if (null != spanBos) {
							for (SpanBo spanBo : spanBos) {
								System.out.println("########" + spanBo.getRpc());
								count(spanBo.getRpc(), map);
								long incrementAndGet = count.incrementAndGet();
//								if(count.incrementAndGet() > 10000){
//									return null;
//								}
							}
						}
					} catch (Exception e) {
						e.printStackTrace();
					}
					return null;
				}
			});
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private void createIndexScan() {
		RowKeyDistributorByHashPrefix.OneByteSimpleHash oneByteSimpleHash = new RowKeyDistributorByHashPrefix.OneByteSimpleHash(32);
		RowKeyDistributorByHashPrefix rowKeyDistributorByHashPrefix = new RowKeyDistributorByHashPrefix(oneByteSimpleHash);
		HbaseApplicationTraceIndexDao hbaseApplicationTraceIndexDao = new HbaseApplicationTraceIndexDao();
		hbaseApplicationTraceIndexDao.setHbaseOperations2(template);
		hbaseApplicationTraceIndexDao.setTraceIdRowKeyDistributor(rowKeyDistributorByHashPrefix);
		final long start = DateUtils.parse("2017-02-20 09:00:00", "yyyy-MM-dd HH:mm:ss").getTime();
		final long end = DateUtils.parse("2017-02-20 18:00:00", "yyyy-MM-dd HH:mm:ss").getTime();
		final Range range = Range.createUncheckedRange(start, end);
		List<Dot> dots = hbaseApplicationTraceIndexDao.scanTraceScatterData("ucenter-pro-uic", range, 5000);
		if(null != dots){
			for(Dot d: dots){
				System.out.println("###########"+d.getTransactionIdAsString());
			}
		}
	}
}
